package com.adcc.monitoragent.mq;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class MqLink {
	private final static Logger log = LoggerFactory.getLogger(MqLink.class);
	/**
	 * 连接工厂
	 */
	private ActiveMQConnectionFactory connectionFactory = null;
	/**
	 * 连接
	 */
	private Connection connection = null;
	/**
	 * 生产者session
	 */
	private Session producerSession = null;
	/**
	 * 生产者
	 */
	private MessageProducer producer = null;
	/**
	 * 消费者session
	 */
	private Session consumerSession = null;
	/**
	 * 消费者
	 */
	private MessageConsumer consumer = null;
	/**
	 * 连接地址
	 */
	private String serverAddr;
	/**
	 * 队列名称
	 */
	private String queueName = "monitoragent-test-alive";
	/**
	 * 消息
	 */
	private TextMessage message = null;
	/**
	 * 是否连接
	 */
	private boolean connected = false;

	private boolean linkOk = false;

	public MqLink(String serverAddr) {
		this.serverAddr = serverAddr;
	}

	/**
	 * 初始化连接
	 * 
	 * @throws Exception
	 */
	private void initialize() {
		try {
			connectionFactory = new ActiveMQConnectionFactory(this.serverAddr);
			connection = connectionFactory.createConnection(
					ActiveMQConnection.DEFAULT_USER,
					ActiveMQConnection.DEFAULT_PASSWORD);
			connection.start();
			producerSession = connection.createSession(false,
					Session.AUTO_ACKNOWLEDGE);
			Queue destination = producerSession.createQueue(this.queueName);
			producer = producerSession.createProducer(destination);
			producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
			producer.setTimeToLive(120000);
			message = producerSession.createTextMessage();
			connected = true;
		} catch (Exception ex) {
			log.error("mq连接失败" + serverAddr);
			cleanup();
		}
	}

	/**
	 * 关闭连接
	 */
	public void cleanup() {
		if (consumer != null) {
			try {
				consumer.close();
				consumer = null;
			} catch (JMSException ex) {
				log.error("消费者关闭失败：" + ex + serverAddr);
			}
		}

		if (consumerSession != null) {
			try {
				consumerSession.close();
				consumerSession = null;
			} catch (JMSException ex) {
				log.error("会话关闭失败：" + ex + serverAddr);
			}
		}

		if (producer != null) {
			try {
				producer.close();
				producer = null;
			} catch (Exception ex) {
				log.error("生产者关闭失败：" + ex + serverAddr);
			}
		}
		if (producerSession != null) {
			try {
				producerSession.close();
				producerSession = null;
			} catch (Exception ex) {
				log.error("session关闭失败：" + ex + serverAddr);
			}
		}
		if (connection != null) {
			try {
				connection.close();
				connection = null;
			} catch (JMSException ex) {
				log.error("连接关闭失败：" + ex + serverAddr);
			}
		}
		connected = false;
	}

	/**
	 * 发送
	 * 
	 * @param object
	 */
	public boolean send(String object) {
		if (!connected) {
			initialize();
		}

		if (connected) {
			try {
				message.setText(object);
				producer.send(message);
				log.debug(serverAddr + "已发送：" + object);
				return true;
			} catch (JMSException ex) {
				log.error(serverAddr + "发送失败：" + ex);
				cleanup();
			}
		} else {
			log.error("初始化失败导致发送失败" + serverAddr);
		}
		return false;
	}

	public void startRecive() {
		if (!connected) {
			initialize();
		}

		if (connected && consumerSession == null) {
			try {
				consumerSession = connection.createSession(false,
						Session.DUPS_OK_ACKNOWLEDGE);
				Queue destination = consumerSession.createQueue(queueName);
				consumer = consumerSession.createConsumer(destination);
				consumer.setMessageListener(new MessageListener() {
					@Override
					public void onMessage(Message arg0) {
						// 接收到数据处理
						linkOk = true;
					}
				});
			} catch (Exception e) {
				log.error("接收数据失败" + serverAddr);
			}

		}
	}
	
	/**
	 * 连接测试
	 * @return
	 */
	public boolean doTest() {
		this.setLinkOk(false);
		for (int j = 0; j < 5; j++) {// 循环发送5次消息
			if (!this.send("t")) {
				return false;
			}
		}
		this.startRecive();// 启动消息接收
		for (int j = 0; j < 200; j++) {// 一秒没有回复超时
			try {
				Thread.sleep(5);
			} catch (InterruptedException e) {
			}
			if (this.isLinkOk()) {
				return true;//消息成功回复
			}
		}
		return false;
	}

	public boolean isLinkOk() {
		return linkOk;
	}

	public void setLinkOk(boolean linkOk) {
		this.linkOk = linkOk;
	}

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		MqLink link = new MqLink(
				"failover:(tcp://127.0.0.1:61616)?initialReconnectDelay=100&timeout=3000&jms.useAsyncSend=true");
		for (int i = 0; i < 1000; i++) {
			link.doTest();
			System.out.println(link.isLinkOk());
			try {
				Thread.sleep(2000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		
	}

}
